package com.etl.component.basic;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.common.collect.Lists;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.param.TableColumn;
import com.component.api.model.param.ParamPair;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.List;
import java.util.Map;

/**
 * 描述：
 * 测试案例
 *   实现简单逻辑 单列加1
 * @author xianggj
 * @Date 2021/11/1 17:08
 **/
public class DataTestComponent extends AbstractFlinkComponent {
    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        System.out.println(getRule() + "开始执行了");
        TimeInterval timer = DateUtil.timer();
        Map<String, Object> paramMap = paramToObject(paramPairs);
        TableColumn column = (TableColumn)paramMap.get("column");
        String numStr = (String)paramMap.get("num");
        Integer num = getInteger(numStr);
        SetData input = (SetData)datas.get(0).getValue();
        List<String> colName = input.getColName();
        final int i = colName.indexOf(column.getColumnName());
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();

        RowTypeInfo rowTypeInfo = changeType(i, dataset);
        //安装第6列元素进行分组 然后拿0列最大值替换 第5列取最小值 必须是int类型
//        dataset = dataset.map(e->{
//            Integer integer = getInteger(e.getField(i));
//            e.setField(i, Integer.valueOf(integer + num));
//            return e;
//        }).groupBy(6).aggregate(Aggregations.MAX, 0);
//                .and(Aggregations.MAX,5);
        String callLocation = Utils.getCallLocationName();
        dataset = new MapOperator<Row, Row>(dataset, rowTypeInfo, e ->{
            Integer integer = getInteger(e.getField(i));
            e.setField(i, Integer.valueOf(integer + num));
            return e;
        }, callLocation);

        System.out.println(getRule() + "执行完成了");
        ProcessResult processResult = new ProcessResult();
        processResult.addAbstractTableData("output",
                new SetData(dataset, input.getColumns()));
        System.out.println(getRule() + "耗时：" + timer.interval()+"ms");
        return processResult;
    }

    /**
     * 将某一列强行转换为int类型
     * @param i
     * @param dataset
     * @return
     */
    private RowTypeInfo changeType(int i, DataSet<Row> dataset) {
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        TypeInformation<?>[] fieldTypes = type.getFieldTypes();
        fieldTypes[i] = BasicTypeInfo.INT_TYPE_INFO;
        return new RowTypeInfo(fieldTypes, type.getFieldNames());
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        SetData input = (SetData)datas.get(0).getValue();
        List<TableColumn> columns = input.getColumns();
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @Override
    public String getRule() {
        return "data_test";
    }
}
